package com.gizwits.kafkaStream

import com.gizwits.util.CaseClasssDomain.User
import com.gizwits.util._
import kafka.serializer.StringDecoder
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object ConsumerApplication {

  def main(args: Array[String]) {
    val sparkConf = new SparkConf().setAppName("KafkaActor")
    val sc = new SparkContext(sparkConf)
    val ssc = new StreamingContext(sc, Seconds(1))
    ssc.checkpoint(".")

    val kafkaParams = Map(
      "zookeeper.connect" ->Constants.ZKQUORUM,
      "group.id" ->Constants.GROUP,
      "auto.commit.interval.ms"-> "10",
      "auto.offset.reset" -> "smallest")

    val topics = Map(
      Constants.KafkaTopic -> 1
    )
    val stream  = KafkaUtils.createStream[String,User,StringDecoder,MsgCommandDecoder[User]](ssc,kafkaParams,Map(Constants.KafkaTopic -> 1),StorageLevel.MEMORY_ONLY)

    stream.print()
    ssc.start()
    ssc.awaitTermination()

  }
}
